package com.stqf.scala.flink

import cn.hutool.core.util.RandomUtil

import java.time.{Instant, LocalDateTime, ZoneId}
import java.time.format.DateTimeFormatter
import java.util.concurrent.TimeUnit
import java.util.{Objects, UUID}
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.common.state.ReducingStateDescriptor
import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton
import org.apache.flink.api.common.typeutils.{SimpleTypeSerializerSnapshot, TypeSerializerSnapshot}
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.core.memory.{DataInputView, DataOutputView}
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.assigners._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.{TimeWindow, Window}
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import scala.beans.BeanProperty

class Payload {

  @BeanProperty var kind: String = _

  @BeanProperty var faceId: String = _

  @BeanProperty var time: String = _

  @BeanProperty var ts: Long = _

  override def toString: String = s"Payload[faceId=$faceId, ts=($ts $time), kind=$kind]"

}

class PayloadFactory extends RichSourceFunction[Payload] {

  private var isStop: Boolean = false

  private val kinds: Array[String] = Array(null, "snap", "recon", "alarm", "detect", "partner")

  override def run(sourceContext: SourceFunction.SourceContext[Payload]): Unit = {
    while (!isStop) {
      val kindId = RandomUtil.randomInt(kinds.length)
      val faceId = UUID.randomUUID().toString
      val ts = System.currentTimeMillis()

      val payloadItem = new Payload()
      payloadItem.setTs(ts)
      payloadItem.setFaceId(faceId)
      payloadItem.setKind(kinds.apply(kindId))

      sourceContext.collect(payloadItem)

      val time = RandomUtil.randomInt(2000) + 2000
      TimeUnit.MILLISECONDS.sleep(time)

    }
  }

  override def cancel(): Unit = {
    isStop = true
  }

}

class LongSerializer extends TypeSerializerSingleton[Long] {
  val instance: LongSerializer = new LongSerializer()

  override def isImmutableType: Boolean = true

  override def createInstance(): Long = 0L

  override def copy(from: Long): Long = from

  override def copy(from: Long, reuse: Long): Long = from

  override def getLength: Int = 8

  override def serialize(record: Long, target: DataOutputView): Unit = target.writeLong(record)

  override def deserialize(source: DataInputView): Long = source.readLong

  override def deserialize(reuse: Long, source: DataInputView): Long = deserialize(source)

  override def copy(source: DataInputView, target: DataOutputView): Unit = target.writeLong(source.readLong)

  override def snapshotConfiguration(): TypeSerializerSnapshot[Long] = new SimpleTypeSerializerSnapshot[Long](() => instance) {}

}

class TimeOrCountTrigger[W <: Window] extends Trigger[AnyRef, W] {

  val log: Logger = LoggerFactory.getLogger(classOf[TimeOrCountTrigger[W]])

  var size: Long = _

  var interval: Long = _

  val countStateDesc: ReducingStateDescriptor[Long] = new ReducingStateDescriptor("count", new ReduceFunction[Long] {
    override def reduce(o1: Long, o2: Long): Long = o1 + o2
  }, classOf[Long])

  val timeStateDesc: ReducingStateDescriptor[Long] = new ReducingStateDescriptor("fire-interval", new ReduceFunction[Long] {
    override def reduce(o1: Long, o2: Long): Long = if (o1 > o2) o2 else o1
  }, classOf[Long])

  def this(count: Long, time: Time) = {
    this()
    this.size = count
    this.interval = time.toMilliseconds
  }

  override def onElement(element: AnyRef, timestamp: Long, window: W, ctx: Trigger.TriggerContext): TriggerResult = {
    //log.debug("onElement ... [{}] ... invoke ", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")))

    val countState = ctx.getPartitionedState(countStateDesc)
    countState.add(1)

    val timeState = ctx.getPartitionedState(timeStateDesc)

    if (countState.get() >= size) {
      log.debug("TimeOrCountTrigger[Count] triggered, count:{} TriggerResult:FIRE", countState.get())

      countState.clear()
      if (timeState.get() != window.maxTimestamp()) {
        log.debug("TimeOrCountTrigger delete, currentTs:{} maxTs:{}", timeState.get(), window.maxTimestamp())
        ctx.deleteProcessingTimeTimer(timeState.get())
      }

      timeState.clear()

      return TriggerResult.FIRE
    }

    val ts = ctx.getCurrentProcessingTime
    if (Objects.isNull(timeState.get())) {
      val nextFireTimestamp: Long = ts + interval
      ctx.registerProcessingTimeTimer(nextFireTimestamp)
      timeState.add(nextFireTimestamp)
    }

    TriggerResult.CONTINUE
  }

  override def onProcessingTime(time: Long, window: W, ctx: Trigger.TriggerContext): TriggerResult = {

    //log.debug("onProcessingTime ... [{}] ... invoke ", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")))

    val countState = ctx.getPartitionedState(countStateDesc)
    val timeState = ctx.getPartitionedState(timeStateDesc)

    if (time == window.maxTimestamp) {
      //log.info("window close : {}", time)
      log.debug("TimeOrCountTrigger[Time] triggered, time:{} TriggerResult:FIRE_AND_PURGE", time)
      countState.clear()
      if (Objects.nonNull(timeState.get())) {
        ctx.deleteProcessingTimeTimer(timeState.get)
        timeState.clear()
      }
      return TriggerResult.FIRE_AND_PURGE
    }

    if (Objects.nonNull(timeState.get()) && timeState.get == time) {
      log.debug("TimeOrCountTrigger[Time] triggered, time:{} TriggerResult:FIRE", time)
      countState.clear()
      timeState.clear()
      return TriggerResult.FIRE
    }

    TriggerResult.CONTINUE
  }

  override def onEventTime(time: Long, window: W, ctx: Trigger.TriggerContext): TriggerResult = {
    log.debug("onEventTime ... [{}] ... invoke ", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")))

    TriggerResult.CONTINUE
  }

  override def clear(window: W, ctx: Trigger.TriggerContext): Unit = {
    //log.debug("clear ... [{}] ... invoke ", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")))
  }

}

/**
 * Created By admin
 * Date: 2020/07/13
 * Description: 流(批)处理探索
 */
object RecvCustomStreaming {

  val log: Logger = LoggerFactory.getLogger(RecvCustomStreaming.getClass)

  val DTF: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")

  implicit val tpMc: TypeInformation[Payload] = TypeInformation.of(new TypeHint[Payload] {})

  implicit val tpApply: TypeInformation[List[Payload]] = TypeInformation.of(new TypeHint[List[Payload]] {})

  def main(args: Array[String]): Unit = {
    log.debug("StreamOrBatch ... start")

    val source = new PayloadFactory()
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)
    env.getConfig.setGlobalJobParameters(ParameterTool.fromSystemProperties())

    val seconds: Long = 15L
    val window = TumblingProcessingTimeWindows.of(Time.seconds(seconds))
    val trigger: TimeOrCountTrigger[Window] = new TimeOrCountTrigger[Window](5, Time.seconds(seconds))

    val res = env.addSource(source).setParallelism(1)
      .map(item => {
        val tsItem = item.getTs
        val timeItem = LocalDateTime.ofInstant(Instant.ofEpochMilli(tsItem), ZoneId.systemDefault()).format(DTF)
        item.setTime(timeItem)
        //log.debug("Item: ... {}", item.ts)
        item
      }).name("SetTime")
      .filter(item => item.kind != null).name("FilterNull")
      .windowAll(window).trigger(trigger).apply(new AllWindowFunction[Payload, List[Payload], TimeWindow]() {
      override def apply(window: TimeWindow, input: Iterable[Payload], out: Collector[List[Payload]]): Unit = {
        val records = input.toList
        val times = records.map(recordItem => recordItem.getTime)
        log.info("Proc: ... size:{} items:{}", records.size, times)
        out.collect(records)
      }
    }).name("TImeOrCountWin")

    //res.print().setParallelism(1)
    res.addSink(payloads => {
      log.info("Sink: ... size:{} items:{}", payloads.size, payloads)
    }).setParallelism(1)


    env.execute("StreamOrBatchLearn")

  }

}
